package com.atbeijing.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming06_Transform {

    def main(args: Array[String]): Unit = {

        // TODO 创建环境对象
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
        val ssc = new StreamingContext(sparkConf, Seconds(3))


        // 从socket中获取的数据是一行一行的字符串
        val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

        //这是一个无状态转化操作
        val wordsDS: DStream[String] = socketDS.flatMap(_.split(" "))

        val wordToOneDS: DStream[(String, Int)] = wordsDS.map((_,1))

        val wordToCountDS = wordToOneDS.reduceByKey(_+_)

        // 离散化流无法排序，可以将DStream转换为RDD进行排序
        // 原语 ： DStream对象的方法
        // 算子 ： RDD对象的方法
        // 方法 ： Scala对象的方法

        // Code Driver （1）
        wordToCountDS.map (
            kv => {
                // Code Executor （Task数量）
                kv._2
            }
        )

        // Code Driver （1）
        wordToCountDS.transform(
            rdd => {
                // Code Driver (周期性执行)
                rdd.map( // 算子
                    kv => {
                        // Code Executor （Task数量）
                        kv._2
                    }
                )
            }
        )

//        wordToCountDS.transform(
//            rdd => {
//                rdd.map(_._2)
//            }
//        ).print()
//
//        wordToCountDS.map(_._2).print

        ssc.start()
        ssc.awaitTermination()
    }
}
